package com.atguigu.market.anaylsis;

import com.atguigu.market.anaylsis.bean.AdClickEvent;
import com.atguigu.market.anaylsis.bean.AdCountViewByProvince;
import com.atguigu.market.anaylsis.bean.BlackListUserWarning;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.net.URL;
import java.sql.Timestamp;

public class AdStatisticsByProvince {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        URL resource = AdStatisticsByProvince.class.getResource("/AdClickLog.csv");
        DataStream<AdClickEvent> dataStream = env.readTextFile(resource.getPath())
                .map(data -> {
                    String[] fields = data.split(",");
                    return new AdClickEvent(new Long(fields[0]), new Long(fields[1]), fields[2], new Long(fields[4]));
                }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<AdClickEvent>() {
                    @Override
                    public long extractAscendingTimestamp(AdClickEvent adClickEvent) {
                        return adClickEvent.getTimestamp() * 1000L;
                    }
                });

        SingleOutputStreamOperator<AdClickEvent> filterDataStream = dataStream.keyBy("userId", "adId")
                .process(new FilterBlankUser(100));

        filterDataStream.print();

        //基于省份分组，开窗聚合
        SingleOutputStreamOperator<AdCountViewByProvince> resultStream = filterDataStream.keyBy(data -> data.getProvince())
                .timeWindow(Time.hours(1), Time.minutes(5))
                .aggregate(new AdCountAgg(), new AdCountResult());

        resultStream.print();
        env.execute("Ad count Stream");


    }

    //自定义过滤函数
    public static class FilterBlankUser extends KeyedProcessFunction<Tuple, AdClickEvent, AdClickEvent> {
        private Integer countUpperBound;

        public FilterBlankUser(Integer countUpperBound) {
            this.countUpperBound = countUpperBound;
        }

        //定义某个属性点击的上限
        ValueState<Long> countState;
        //定义标识状态，保存当前用户是否已经被发送到了黑名单里面即侧输出流
        ValueState<Boolean> isSentState;

        @Override
        public void open(Configuration parameters) throws Exception {
            countState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("ad-count", Long.class, 0L));
            isSentState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-sent", Boolean.class, false));
        }

        @Override
        public void processElement(AdClickEvent adClickEvent, Context context, Collector<AdClickEvent> collector) throws Exception {
            //判断当前用户对同一个广告的点击次数，如果不够上限就count+1正常输出；如果达到上限，直接过滤，并侧输出到黑名单中
            //获取当前的count值
            Long curCount = countState.value();
            //判断是否是第一个数据,如果是则注册一个第二天零点的计时器
            if (curCount == null || curCount <= 0) {
                long ts = ((context.timerService().currentProcessingTime() / 24 * 60 * 60 * 1000) + 1) * 24 * 60 * 60 * 1000 - 8 * 60 * 60 * 1000;
                context.timerService().registerProcessingTimeTimer(ts);
            }

            if (curCount >= countUpperBound) {
                // 判断是否输出到黑名单过，如果没有就侧输出到侧输出流
                if (isSentState.value()) {
                    context.output(new OutputTag<BlackListUserWarning>("blacklist") {
                    }, new BlackListUserWarning(adClickEvent.getUserId(), adClickEvent.getAdId(), "Click over" + countUpperBound + "times"));
                }
                return;
            }
            //如果没有返回，点击次数加1更新状态，正常输出到主流
            countState.update(curCount + 1L);
            collector.collect(adClickEvent);

        }

        //定时器触发清除所有状态
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<AdClickEvent> out) throws Exception {
            countState.clear();
            isSentState.clear();
        }
    }

    public static class AdCountAgg implements AggregateFunction<AdClickEvent, Long, Long> {

        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(AdClickEvent value, Long accumulator) {
            return accumulator + 1L;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }

    public static class AdCountResult implements WindowFunction<Long, AdCountViewByProvince, String, TimeWindow> {

        @Override
        public void apply(String s, TimeWindow timeWindow, Iterable<Long> iterable, Collector<AdCountViewByProvince> collector) throws Exception {
            collector.collect(new AdCountViewByProvince(s, new Timestamp(timeWindow.getEnd()).toString(), iterable.iterator().next()));
        }
    }
}
